<!DOCTYPE html><html class="theme-next gemini use-motion" lang="zh-CN"><head><meta name="generator" content="Hexo 3.9.0"><meta charset="UTF-8"><meta http-equiv="X-UA-Compatible" content="IE=edge"><meta name="viewport" content="width=device-width,initial-scale=1,maximum-scale=2"><meta name="theme-color" content="#222"><link rel="apple-touch-icon" sizes="180x180" href="/favicon.ico?v=6.7.0"><link rel="icon" type="image/png" sizes="32x32" href="/favicon.ico?v=6.7.0"><link rel="icon" type="image/png" sizes="16x16" href="/favicon.ico?v=6.7.0"><script id="hexo.configurations">var NexT=window.NexT||{},CONFIG={root:"/",scheme:"Gemini",version:"6.7.0",sidebar:{position:"left",display:"post",offset:12,b2t:!1,scrollpercent:!1,onmobile:!1},fancybox:!0,fastclick:!0,lazyload:!1,tabs:!0,motion:{enable:!0,async:!1,transition:{post_block:"fadeIn",post_header:"slideDownIn",post_body:"slideDownIn",coll_header:"slideLeftIn",sidebar:"slideUpIn"}},algolia:{applicationID:"",apiKey:"",indexName:"",hits:{per_page:10},labels:{input_placeholder:"Search for Posts",hits_empty:"We didn't find any results for the search: ${query}",hits_stats:"${hits} results found in ${time} ms"}}}</script><meta name="description" content="gRPC 是一个优秀的开源 RPC 框架，它能够实现双向流式调用。本文从源码的角度出发，分层剖析 gRPC 流式调用的实现。"><meta name="keywords" content="源码分析,gRPC"><meta property="og:type" content="article"><meta property="og:title" content="gRPC Bidirectional（双向流）源码分析"><meta property="og:url" content="http://moeyui.cn/开发/d1717587.html"><meta property="og:site_name" content="不是很懂"><meta property="og:description" content="gRPC 是一个优秀的开源 RPC 框架，它能够实现双向流式调用。本文从源码的角度出发，分层剖析 gRPC 流式调用的实现。"><meta property="og:locale" content="zh-CN"><meta property="og:image" content="https://s1.ax1x.com/2018/10/13/iN7ZfU.png"><meta property="og:image" content="https://oeoiy7i1f.qnssl.com/gRPC%E5%8F%8C%E5%90%91%E6%B5%81/drawio.svg"><meta property="og:image" content="https://ws3.sinaimg.cn/large/006tNbRwgy1fw5gx3xqspj30ds0843yj.jpg"><meta property="og:image" content="https://ws1.sinaimg.cn/large/006tNbRwgy1fw5h0qf5vvj30et09yjrj.jpg"><meta property="og:image" content="https://ws3.sinaimg.cn/large/006tNbRwgy1fw5h2ljh83j314w06it9j.jpg"><meta property="og:updated_time" content="2018-10-13T11:44:39.820Z"><meta name="twitter:card" content="summary"><meta name="twitter:title" content="gRPC Bidirectional（双向流）源码分析"><meta name="twitter:description" content="gRPC 是一个优秀的开源 RPC 框架，它能够实现双向流式调用。本文从源码的角度出发，分层剖析 gRPC 流式调用的实现。"><meta name="twitter:image" content="https://s1.ax1x.com/2018/10/13/iN7ZfU.png"><link rel="alternate" href="/atom.xml" title="不是很懂" type="application/atom+xml"><link rel="canonical" href="http://moeyui.cn/开发/d1717587.html"><script id="page.configurations">CONFIG.page={sidebar:""}</script><title>gRPC Bidirectional（双向流）源码分析 | 不是很懂</title><script async="" src="//www.googletagmanager.com/gtag/js?id=UA-75483578-1"></script><script>function gtag(){dataLayer.push(arguments)}window.dataLayer=window.dataLayer||[],gtag("js",new Date),gtag("config","UA-75483578-1")</script><noscript><style>.sidebar-inner,.use-motion .brand,.use-motion .collection-title,.use-motion .comments,.use-motion .menu-item,.use-motion .motion-element,.use-motion .pagination,.use-motion .post-block,.use-motion .post-body,.use-motion .post-header{opacity:initial}.use-motion .logo,.use-motion .site-subtitle,.use-motion .site-title{opacity:initial;top:initial}.use-motion .logo-line-before i{left:initial}.use-motion .logo-line-after i{right:initial}</style></noscript><script>function loadCss(e){var t=document,s=t.head,n=t.createElement("link");n.rel="stylesheet",n.href=e,function e(s){if(t.body)return s();setTimeout(function(){e(s)})}(function(){s.appendChild(n)})}loadCss("/style.css"),loadCss("https://cdn.staticfile.org/font-awesome/4.6.2/css/font-awesome.min.css")</script><noscript><link rel="stylesheet" href="/style.css"><link rel="stylesheet" href="https://cdn.staticfile.org/font-awesome/4.6.2/css/font-awesome.min.css"></noscript></head><body itemscope="" itemtype="http://schema.org/WebPage" lang="zh-CN"><div class="container sidebar-position-left page-post-detail"><div class="headband"></div><header id="header" class="header" itemscope="" itemtype="http://schema.org/WPHeader"><div class="header-inner"><div class="site-brand-wrapper"><div class="site-meta"><div class="custom-logo-site-title"><a href="/" class="brand" rel="start"><span class="logo-line-before"><i></i></span> <span class="site-title">不是很懂</span><span class="logo-line-after"><i></i></span></a></div><h1 class="site-subtitle" itemprop="description">不是很懂你们程序员</h1></div><div class="site-nav-toggle"> <button aria-label="切换导航栏"><span class="btn-bar"></span><span class="btn-bar"></span><span class="btn-bar"></span></button></div></div><nav class="site-nav"><ul id="menu" class="menu"><li class="menu-item menu-item-home"><a href="/" rel="section"><i class="menu-item-icon fa fa-fw fa-home"></i><br>首页</a></li><li class="menu-item menu-item-about"><a href="/about/" rel="section"><i class="menu-item-icon fa fa-fw fa-user"></i><br>关于</a></li><li class="menu-item menu-item-tags"><a href="/tags/" rel="section"><i class="menu-item-icon fa fa-fw fa-tags"></i><br>标签</a></li><li class="menu-item menu-item-categories"><a href="/categories/" rel="section"><i class="menu-item-icon fa fa-fw fa-th"></i><br>分类</a></li><li class="menu-item menu-item-archives"><a href="/archives/" rel="section"><i class="menu-item-icon fa fa-fw fa-archive"></i><br>归档</a></li></ul></nav></div></header> <a href="https://github.com/moeyui1" class="github-corner" title="Follow me on GitHub" aria-label="Follow me on GitHub" rel="noopener" target="_blank"><svg width="80" height="80" viewBox="0 0 250 250" style="fill:#222;color:#fff;position:absolute;top:0;border:0;right:0" aria-hidden="true"><path d="M0 0 115 115 130 115 142 142 250 250 250 0Z"></path><path d="M128.3 109C113.8 99.7 119 89.6 119 89.6 122 82.7 120.5 78.6 120.5 78.6 119.2 72 123.4 76.3 123.4 76.3 127.3 80.9 125.5 87.3 125.5 87.3 122.9 97.6 130.6 101.9 134.4 103.2" fill="currentColor" style="transform-origin:130px 106px" class="octo-arm"></path><path d="M115 115C114.9 115.1 118.7 116.5 119.8 115.4L133.7 101.6C136.9 99.2 139.9 98.4 142.2 98.6 133.8 88 127.5 74.4 143.8 58 148.5 53.4 154 51.2 159.7 51 160.3 49.4 163.2 43.6 171.4 40.1 171.4 40.1 176.1 42.5 178.8 56.2 183.1 58.6 187.2 61.8 190.9 65.4 194.5 69 197.7 73.2 200.1 77.6 213.8 80.2 216.3 84.9 216.3 84.9 212.7 93.1 206.9 96 205.4 96.6 205.1 102.4 203 107.8 198.3 112.5 181.9 128.9 168.3 122.5 157.7 114.1 157.9 116.9 156.7 120.9 152.7 124.9L141 136.5C139.8 137.7 141.6 141.9 141.8 141.8Z" fill="currentColor" class="octo-body"></path></svg></a><main id="main" class="main"><div class="main-inner"><div class="content-wrap"><div id="content" class="content"><div id="posts" class="posts-expand"><article class="post post-type-normal" itemscope="" itemtype="http://schema.org/Article"><div class="post-block"><link itemprop="mainEntityOfPage" href="http://moeyui.cn/开发/d1717587.html"><span hidden="" itemprop="author" itemscope="" itemtype="http://schema.org/Person"><meta itemprop="name" content="moeyui"><meta itemprop="description" content="moeyui | moeyui's blog"><meta itemprop="image" content="/images/blogAvatar.jpg"></span><span hidden="" itemprop="publisher" itemscope="" itemtype="http://schema.org/Organization"><meta itemprop="name" content="不是很懂"></span><header class="post-header"><h2 class="post-title" itemprop="name headline">gRPC Bidirectional（双向流）源码分析</h2><div class="post-meta"><span class="post-time"><span class="post-meta-item-icon"><i class="fa fa-calendar-o"></i></span> <span class="post-meta-item-text">发表于</span> <time title="创建时间：2018-10-13 19:26:28 / 修改时间：19:44:39" itemprop="dateCreated datePublished" datetime="2018-10-13T19:26:28+08:00">2018-10-13</time></span> <span class="post-category"><span class="post-meta-divider">|</span><span class="post-meta-item-icon"><i class="fa fa-folder-o"></i></span> <span class="post-meta-item-text">分类于</span> <span itemprop="about" itemscope="" itemtype="http://schema.org/Thing"><a href="/categories/开发/" itemprop="url" rel="index"><span itemprop="name">开发</span></a></span></span> <span class="post-comments-count"><span class="post-meta-divider">|</span><span class="post-meta-item-icon"><i class="fa fa-comment-o"></i></span> <a href="/开发/d1717587.html#comments" itemprop="discussionUrl"><span class="post-meta-item-text">评论数：</span><span class="post-comments-count valine-comment-count" data-xid="/开发/d1717587.html" itemprop="commentCount"></span></a></span> <span id="/开发/d1717587.html" class="leancloud_visitors" data-flag-title="gRPC Bidirectional（双向流）源码分析"><span class="post-meta-divider">|</span><span class="post-meta-item-icon"><i class="fa fa-eye"></i></span> <span class="post-meta-item-text">阅读次数：</span><span class="leancloud-visitors-count"></span></span></div></header><div class="post-body" itemprop="articleBody"><p>gRPC 是一个优秀的开源 RPC 框架，它能够实现双向流式调用。本文从源码的角度出发，分层剖析 gRPC 流式调用的实现。</p><p><img src="https://s1.ax1x.com/2018/10/13/iN7ZfU.png" alt="iN7ZfU.png"></p><a id="more"></a><h1 id="Overview"><a href="#Overview" class="headerlink" title="Overview"></a>Overview</h1><p>从高层上看，gRPC 可分为三层: <strong>Stub/桩</strong>, <strong>Channel/通道</strong> &amp; <strong>Transport/传输</strong>.</p><h2 id="Stub"><a href="#Stub" class="headerlink" title="Stub"></a>Stub</h2><blockquote><p>The Stub layer is what is exposed to most developers and provides type-safe bindings to whatever datamodel/IDL/interface you are adapting. gRPC comes with a <a href="https://github.com/google/grpc-java/blob/master/compiler" target="_blank" rel="noopener">plugin</a> to the protocol-buffers compiler that generates Stub interfaces out of .protofiles, but bindings to other datamodel/IDL are easy and encouraged.</p></blockquote><h2 id="Channel"><a href="#Channel" class="headerlink" title="Channel"></a>Channel</h2><blockquote><p>The Channel layer is an abstraction over Transport handling that is suitable for interception/decoration and exposes more behavior to the application than the Stub layer. It is intended to be easy for application frameworks to use this layer to address cross-cutting concerns such as logging, monitoring, auth, etc.</p></blockquote><p>Channel 层是 Transport 层处理上的抽象，适合 interception/decoration ，并暴露更多行为给应用（相比 Stub 层）。它的目的是为了使应用框架利用改成方便地实现 address cross-cutting 例如日志、监控、鉴权等。</p><h2 id="Transport"><a href="#Transport" class="headerlink" title="Transport"></a>Transport</h2><p>The Transport layer does the heavy lifting of putting and taking bytes off the wire. The interfaces to it are abstract just enough to allow plugging in of different implementations. Note the transport layer API is considered internal to gRPC and has weaker API guarantees than the core API under package io.grpc.</p><p>gRPC 自带3种 Transport 实现:</p><ol><li>Netty-based transport 是主要的 transport 实现，基于 Netty。为客户端和服务端使用。</li><li>OkHttp-based transport 是一个轻量级的 transport，基于 OkHttp。这主要被用在 Android 上，只能用在客户端。</li><li>In-Process transport 是为服务端和客户端在同一进程时准备的。它对测试很有用，在生产环境也很安全。</li></ol><h1 id="使用"><a href="#使用" class="headerlink" title="使用"></a>使用</h1><p>以 gRPC 官方的 examples 为例看下表层的情况。</p><h2 id="客户端"><a href="#客户端" class="headerlink" title="客户端"></a>客户端</h2><figure class="highlight java"><table><tbody><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">  * Bi-directional example, which can only be asynchronous. Send some chat messages, and print any</span></span><br><span class="line"><span class="comment">  * chat messages that are sent from the server.</span></span><br><span class="line"><span class="comment">  */</span></span><br><span class="line"> <span class="function"><span class="keyword">public</span> CountDownLatch <span class="title">routeChat</span><span class="params">()</span> </span>{</span><br><span class="line">   info(<span class="string">"*** RouteChat"</span>);</span><br><span class="line">   <span class="keyword">final</span> CountDownLatch finishLatch = <span class="keyword">new</span> CountDownLatch(<span class="number">1</span>);</span><br><span class="line">   StreamObserver&lt;RouteNote&gt; requestObserver =</span><br><span class="line">       <span class="comment">// 方法调用还未开始，从这里获得一个 StreamObserver，用于传递请求流</span></span><br><span class="line">       asyncStub.routeChat(<span class="keyword">new</span> StreamObserver&lt;RouteNote&gt;() {</span><br><span class="line">         <span class="comment">// 自实现 StreamObserver，gRPC 会在适当时机调用下面的方法</span></span><br><span class="line">         <span class="meta">@Override</span></span><br><span class="line">         <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onNext</span><span class="params">(RouteNote note)</span> </span>{</span><br><span class="line">           info(<span class="string">"Got message \"{0}\" at {1}, {2}"</span>, note.getMessage(), note.getLocation()</span><br><span class="line">               .getLatitude(), note.getLocation().getLongitude());</span><br><span class="line">         }</span><br><span class="line"></span><br><span class="line">         <span class="meta">@Override</span></span><br><span class="line">         <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onError</span><span class="params">(Throwable t)</span> </span>{</span><br><span class="line">           warning(<span class="string">"RouteChat Failed: {0}"</span>, Status.fromThrowable(t));</span><br><span class="line">           finishLatch.countDown();</span><br><span class="line">         }</span><br><span class="line"></span><br><span class="line">         <span class="meta">@Override</span></span><br><span class="line">         <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onCompleted</span><span class="params">()</span> </span>{</span><br><span class="line">           info(<span class="string">"Finished RouteChat"</span>);</span><br><span class="line">           finishLatch.countDown();</span><br><span class="line">         }</span><br><span class="line">       });</span><br><span class="line"></span><br><span class="line">   <span class="keyword">try</span> {</span><br><span class="line">     RouteNote[] requests =</span><br><span class="line">         {newNote(<span class="string">"First message"</span>, <span class="number">0</span>, <span class="number">0</span>), newNote(<span class="string">"Second message"</span>, <span class="number">0</span>, <span class="number">1</span>),</span><br><span class="line">             newNote(<span class="string">"Third message"</span>, <span class="number">1</span>, <span class="number">0</span>), newNote(<span class="string">"Fourth message"</span>, <span class="number">1</span>, <span class="number">1</span>)};</span><br><span class="line"></span><br><span class="line">     <span class="keyword">for</span> (RouteNote request : requests) {</span><br><span class="line">       info(<span class="string">"Sending message \"{0}\" at {1}, {2}"</span>, request.getMessage(), request.getLocation()</span><br><span class="line">           .getLatitude(), request.getLocation().getLongitude());</span><br><span class="line">       <span class="comment">// 通过 StreamObserver 不断发送请求流</span></span><br><span class="line">       requestObserver.onNext(request);</span><br><span class="line">     }</span><br><span class="line">   } <span class="keyword">catch</span> (RuntimeException e) {</span><br><span class="line">     <span class="comment">// Cancel RPC</span></span><br><span class="line">     requestObserver.onError(e);</span><br><span class="line">     <span class="keyword">throw</span> e;</span><br><span class="line">   }</span><br><span class="line">   <span class="comment">// Mark the end of requests</span></span><br><span class="line">   requestObserver.onCompleted();</span><br><span class="line"></span><br><span class="line">   <span class="comment">// return the latch while receiving happens asynchronously</span></span><br><span class="line">   <span class="keyword">return</span> finishLatch;</span><br><span class="line"> }</span><br></pre></td></tr></tbody></table></figure><h2 id="服务端"><a href="#服务端" class="headerlink" title="服务端"></a>服务端</h2><figure class="highlight java"><table><tbody><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">    * Receives a stream of message/location pairs, and responds with a stream of all previous</span></span><br><span class="line"><span class="comment">    * messages at each of those locations.</span></span><br><span class="line"><span class="comment">    *</span></span><br><span class="line"><span class="comment">    * <span class="doctag">@param</span> responseObserver an observer to receive the stream of previous messages.</span></span><br><span class="line"><span class="comment">    * <span class="doctag">@return</span> an observer to handle requested message/location pairs.</span></span><br><span class="line"><span class="comment">    */</span></span><br><span class="line">   <span class="meta">@Override</span></span><br><span class="line">   <span class="comment">// 传入参数是由 gRPC 生成的一个 StreamObserver，通过它可以实现流式响应</span></span><br><span class="line">   <span class="function"><span class="keyword">public</span> StreamObserver&lt;RouteNote&gt; <span class="title">routeChat</span><span class="params">(<span class="keyword">final</span> StreamObserver&lt;RouteNote&gt; responseObserver)</span> </span>{</span><br><span class="line">     <span class="comment">// 这个 StreamObserver 在适当时机被 gRPC 调用</span></span><br><span class="line">     <span class="keyword">return</span> <span class="keyword">new</span> StreamObserver&lt;RouteNote&gt;() {</span><br><span class="line">       <span class="meta">@Override</span></span><br><span class="line">       <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onNext</span><span class="params">(RouteNote note)</span> </span>{</span><br><span class="line">         List&lt;RouteNote&gt; notes = getOrCreateNotes(note.getLocation());</span><br><span class="line"></span><br><span class="line">         <span class="comment">// Respond with all previous notes at this location.</span></span><br><span class="line">         <span class="keyword">for</span> (RouteNote prevNote : notes.toArray(<span class="keyword">new</span> RouteNote[<span class="number">0</span>])) {</span><br><span class="line">           responseObserver.onNext(prevNote);</span><br><span class="line">         }</span><br><span class="line"></span><br><span class="line">         <span class="comment">// Now add the new note to the list</span></span><br><span class="line">         notes.add(note);</span><br><span class="line">       }</span><br><span class="line"></span><br><span class="line">       <span class="meta">@Override</span></span><br><span class="line">       <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onError</span><span class="params">(Throwable t)</span> </span>{</span><br><span class="line">         logger.log(Level.WARNING, <span class="string">"routeChat cancelled"</span>);</span><br><span class="line">       }</span><br><span class="line"></span><br><span class="line">       <span class="meta">@Override</span></span><br><span class="line">       <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onCompleted</span><span class="params">()</span> </span>{</span><br><span class="line">         responseObserver.onCompleted();</span><br><span class="line">       }</span><br><span class="line">     };</span><br><span class="line">   }</span><br></pre></td></tr></tbody></table></figure><h2 id="StreamObserver"><a href="#StreamObserver" class="headerlink" title="StreamObserver"></a>StreamObserver</h2><p>看了上面的代码可以知道，StreamObserver 正是 gRPC 在 Stub 层提供的一个流式 Observer，通过它可以实现接收和发送流。</p><figure class="highlight java"><table><tbody><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> io.grpc.stub;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * Receives notifications from an observable stream of messages.</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * &lt;p&gt;It is used by both the client stubs and service implementations for sending or receiving</span></span><br><span class="line"><span class="comment"> * stream messages. It is used for all {<span class="doctag">@link</span> io.grpc.MethodDescriptor.MethodType}, including</span></span><br><span class="line"><span class="comment"> * {<span class="doctag">@code</span> UNARY} calls.  For outgoing messages, a {<span class="doctag">@code</span> StreamObserver} is provided by the GRPC</span></span><br><span class="line"><span class="comment"> * library to the application. For incoming messages, the application implements the</span></span><br><span class="line"><span class="comment"> * {<span class="doctag">@code</span> StreamObserver} and passes it to the GRPC library for receiving.</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * &lt;p&gt;Implementations are not required to be thread-safe (but should be</span></span><br><span class="line"><span class="comment"> * &lt;a href="http://www.ibm.com/developerworks/library/j-jtp09263/"&gt;thread-compatible&lt;/a&gt;).</span></span><br><span class="line"><span class="comment"> * Separate {<span class="doctag">@code</span> StreamObserver}s do</span></span><br><span class="line"><span class="comment"> * not need to be synchronized together; incoming and outgoing directions are independent.</span></span><br><span class="line"><span class="comment"> * Since individual {<span class="doctag">@code</span> StreamObserver}s are not thread-safe, if multiple threads will be</span></span><br><span class="line"><span class="comment"> * writing to a {<span class="doctag">@code</span> StreamObserver} concurrently, the application must synchronize calls.</span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">interface</span> <span class="title">StreamObserver</span>&lt;<span class="title">V</span>&gt;  </span>{</span><br><span class="line">  <span class="comment">/**</span></span><br><span class="line"><span class="comment">   * Receives a value from the stream.</span></span><br><span class="line"><span class="comment">   *</span></span><br><span class="line"><span class="comment">   * &lt;p&gt;Can be called many times but is never called after {<span class="doctag">@link</span> #onError(Throwable)} or {<span class="doctag">@link</span></span></span><br><span class="line"><span class="comment">   * #onCompleted()} are called.</span></span><br><span class="line"><span class="comment">   *</span></span><br><span class="line"><span class="comment">   * &lt;p&gt;Unary calls must invoke onNext at most once.  Clients may invoke onNext at most once for</span></span><br><span class="line"><span class="comment">   * server streaming calls, but may receive many onNext callbacks.  Servers may invoke onNext at</span></span><br><span class="line"><span class="comment">   * most once for client streaming calls, but may receive many onNext callbacks.</span></span><br><span class="line"><span class="comment">   *</span></span><br><span class="line"><span class="comment">   * &lt;p&gt;If an exception is thrown by an implementation the caller is expected to terminate the</span></span><br><span class="line"><span class="comment">   * stream by calling {<span class="doctag">@link</span> #onError(Throwable)} with the caught exception prior to</span></span><br><span class="line"><span class="comment">   * propagating it.</span></span><br><span class="line"><span class="comment">   *</span></span><br><span class="line"><span class="comment">   * <span class="doctag">@param</span> value the value passed to the stream</span></span><br><span class="line"><span class="comment">   */</span></span><br><span class="line">  <span class="function"><span class="keyword">void</span> <span class="title">onNext</span><span class="params">(V value)</span></span>;</span><br><span class="line"></span><br><span class="line">  <span class="comment">/**</span></span><br><span class="line"><span class="comment">   * Receives a terminating error from the stream.</span></span><br><span class="line"><span class="comment">   *</span></span><br><span class="line"><span class="comment">   * &lt;p&gt;May only be called once and if called it must be the last method called. In particular if an</span></span><br><span class="line"><span class="comment">   * exception is thrown by an implementation of {<span class="doctag">@code</span> onError} no further calls to any method are</span></span><br><span class="line"><span class="comment">   * allowed.</span></span><br><span class="line"><span class="comment">   *</span></span><br><span class="line"><span class="comment">   * &lt;p&gt;{<span class="doctag">@code</span> t} should be a {<span class="doctag">@link</span> io.grpc.StatusException} or {<span class="doctag">@link</span></span></span><br><span class="line"><span class="comment">   * io.grpc.StatusRuntimeException}, but other {<span class="doctag">@code</span> Throwable} types are possible. Callers should</span></span><br><span class="line"><span class="comment">   * generally convert from a {<span class="doctag">@link</span> io.grpc.Status} via {<span class="doctag">@link</span> io.grpc.Status#asException()} or</span></span><br><span class="line"><span class="comment">   * {<span class="doctag">@link</span> io.grpc.Status#asRuntimeException()}. Implementations should generally convert to a</span></span><br><span class="line"><span class="comment">   * {<span class="doctag">@code</span> Status} via {<span class="doctag">@link</span> io.grpc.Status#fromThrowable(Throwable)}.</span></span><br><span class="line"><span class="comment">   *</span></span><br><span class="line"><span class="comment">   * <span class="doctag">@param</span> t the error occurred on the stream</span></span><br><span class="line"><span class="comment">   */</span></span><br><span class="line">  <span class="function"><span class="keyword">void</span> <span class="title">onError</span><span class="params">(Throwable t)</span></span>;</span><br><span class="line"></span><br><span class="line">  <span class="comment">/**</span></span><br><span class="line"><span class="comment">   * Receives a notification of successful stream completion.</span></span><br><span class="line"><span class="comment">   *</span></span><br><span class="line"><span class="comment">   * &lt;p&gt;May only be called once and if called it must be the last method called. In particular if an</span></span><br><span class="line"><span class="comment">   * exception is thrown by an implementation of {<span class="doctag">@code</span> onCompleted} no further calls to any method</span></span><br><span class="line"><span class="comment">   * are allowed.</span></span><br><span class="line"><span class="comment">   */</span></span><br><span class="line">  <span class="function"><span class="keyword">void</span> <span class="title">onCompleted</span><span class="params">()</span></span>;</span><br><span class="line">}</span><br></pre></td></tr></tbody></table></figure><h1 id="简图"><a href="#简图" class="headerlink" title="简图"></a>简图</h1><p>整个实现是建立在复杂的监听模式基础上的。以 Client 端为视角：</p><p><img src="https://oeoiy7i1f.qnssl.com/gRPC%E5%8F%8C%E5%90%91%E6%B5%81/drawio.svg" alt="drawio"></p><p>Server 端视角可触类旁通。</p><h1 id="分层分析"><a href="#分层分析" class="headerlink" title="分层分析"></a>分层分析</h1><p>上面的一图比较多比较乱，下面来逐层分析。</p><h2 id="Stub-层"><a href="#Stub-层" class="headerlink" title="Stub 层"></a>Stub 层</h2><p>这一层关注 <code>StreamObserver</code>，它的代码上文已经贴过了。使用者就是通过它实现流式通信。结合上文的源码分析，可以得到以下结论：</p><ul><li>对于客户端来说，resp 由 gRPC 生成并返回，req 则是客户端自行实现 <code>StreamObserver</code>。</li><li>对于服务端来说，req 由 gRPC 生成并作为入参交给服务端方法，resp 则是服务端自己实现并返回给 gRPC。</li></ul><p>模糊的地方就在 gRPC 如何生成一个 <code>StreamObserver</code>。以客户端为例分析：</p><p><img src="https://ws3.sinaimg.cn/large/006tNbRwgy1fw5gx3xqspj30ds0843yj.jpg" alt=""></p><p>发起请求时，首先通过 Channel 获得一个 <code>ClientCall</code>，这个 call 是 Channel 层的，在 Stub 层客户端需要使用 <code>StreamObserver</code>，故使用了一个 <code>CallToStreamObserverAdapter</code> 来将 call 包起来返回给客户端。</p><figure class="highlight java"><table><tbody><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> <span class="class"><span class="keyword">class</span> <span class="title">CallToStreamObserverAdapter</span>&lt;<span class="title">T</span>&gt; <span class="keyword">extends</span> <span class="title">ClientCallStreamObserver</span>&lt;<span class="title">T</span>&gt; </span>{</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> frozen;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">final</span> ClientCall&lt;T, ?&gt; call;</span><br><span class="line">    <span class="keyword">private</span> Runnable onReadyHandler;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> autoFlowControlEnabled = <span class="keyword">true</span>;</span><br><span class="line"></span><br><span class="line">    <span class="comment">// Non private to avoid synthetic class</span></span><br><span class="line">    CallToStreamObserverAdapter(ClientCall&lt;T, ?&gt; call) {</span><br><span class="line">      <span class="keyword">this</span>.call = call;</span><br><span class="line">    }</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">freeze</span><span class="params">()</span> </span>{</span><br><span class="line">      <span class="keyword">this</span>.frozen = <span class="keyword">true</span>;</span><br><span class="line">    }</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onNext</span><span class="params">(T value)</span> </span>{</span><br><span class="line">      call.sendMessage(value);</span><br><span class="line">    }</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onError</span><span class="params">(Throwable t)</span> </span>{</span><br><span class="line">      call.cancel(<span class="string">"Cancelled by client with StreamObserver.onError()"</span>, t);</span><br><span class="line">    }</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onCompleted</span><span class="params">()</span> </span>{</span><br><span class="line">      call.halfClose();</span><br><span class="line">    }</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">boolean</span> <span class="title">isReady</span><span class="params">()</span> </span>{</span><br><span class="line">      <span class="keyword">return</span> call.isReady();</span><br><span class="line">    }</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setOnReadyHandler</span><span class="params">(Runnable onReadyHandler)</span> </span>{</span><br><span class="line">      <span class="keyword">if</span> (frozen) {</span><br><span class="line">        <span class="keyword">throw</span> <span class="keyword">new</span> IllegalStateException(<span class="string">"Cannot alter onReadyHandler after call started"</span>);</span><br><span class="line">      }</span><br><span class="line">      <span class="keyword">this</span>.onReadyHandler = onReadyHandler;</span><br><span class="line">    }</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">disableAutoInboundFlowControl</span><span class="params">()</span> </span>{</span><br><span class="line">      <span class="keyword">if</span> (frozen) {</span><br><span class="line">        <span class="keyword">throw</span> <span class="keyword">new</span> IllegalStateException(<span class="string">"Cannot disable auto flow control call started"</span>);</span><br><span class="line">      }</span><br><span class="line">      autoFlowControlEnabled = <span class="keyword">false</span>;</span><br><span class="line">    }</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">request</span><span class="params">(<span class="keyword">int</span> count)</span> </span>{</span><br><span class="line">      call.request(count);</span><br><span class="line">    }</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setMessageCompression</span><span class="params">(<span class="keyword">boolean</span> enable)</span> </span>{</span><br><span class="line">      call.setMessageCompression(enable);</span><br><span class="line">    }</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">cancel</span><span class="params">(@Nullable String message, @Nullable Throwable cause)</span> </span>{</span><br><span class="line">      call.cancel(message, cause);</span><br><span class="line">    }</span><br><span class="line">  }</span><br></pre></td></tr></tbody></table></figure><p>看了 <code>CallToStreamObserverAdapter</code> 的源码，就知道客户端在调用 <code>StreamObserver.next(value)</code> 方法时，实际就是调用了 <code>call.sendMessage(value)</code> 发送消息。其实 <code>CallToStreamObserverAdapter</code> 就是 <code>ClientCall</code> 在 Stub 层的适配器。</p><p>上面说的都是请求的 <code>StreamObserver</code>，那响应的 <code>StreamObserver</code> 呢？因为客户端已经在自实现的 <code>StreamObserver</code> 中实现了对响应的处理方法，所以客户端后续已经不需要与响应的 <code>StreamObserver</code> 交互了，所以这个自实现的 <code>StreamObserver</code> 直接被传到了 Channel 层。</p><h2 id="Channel-层"><a href="#Channel-层" class="headerlink" title="Channel 层"></a>Channel 层</h2><p>这一层逻辑开始复杂，也是本文主要关注的层级。这层主要关注一个 ClientCallImpl 和一个 StreamObserverToCallListenerAdapter。</p><p><img src="https://ws1.sinaimg.cn/large/006tNbRwgy1fw5h0qf5vvj30et09yjrj.jpg" alt=""></p><p>先看 ClientCallImpl，它是 ClientCall 的实现类，它内部持有一个 ClientCall.Listener，这是用来监听什么的呢？</p><p>上面 Stub 层末尾讲到响应的 StreamObserver 传到了 Channel 层，实际就是到了 StreamObserverToCallListenerAdapter 的 observer 中。<em>gRPC 的命名都很直白。</em>而其中的 adapter 就是 stub 层的 CallToStreamObserverAdapter。它持有两个 StreamObserver 想干啥？</p><p>实际上 StreamObserverToCallListenerAdapter 接管了两个 StreamObserver 的监听，合并成一个 ClientCall.Listener 去监听 ClientCallImpl，到这里，自定义的监听终于和请求绑定在一起了。</p><p>ClientCallImpl 内部封装了诸如消息发送等网络细节，通过它持有的 ClientStream 类型引用实现。这是 Transport 层的概念了。</p><h2 id="Transport-层"><a href="#Transport-层" class="headerlink" title="Transport 层"></a>Transport 层</h2><p>这一层就更复杂些，可对接多种实现，本文不做讨论<del>（偷懒了</del>。</p><h2 id="协议层"><a href="#协议层" class="headerlink" title="协议层"></a>协议层</h2><p>gRPC 基于 Http2，多路复用是 Http2 的一大特性。这一特性得益于 frame 的设计，<strong>frame 的 Header 中标识了它属于的流</strong>。</p><p><img src="https://ws3.sinaimg.cn/large/006tNbRwgy1fw5h2ljh83j314w06it9j.jpg" alt=""></p><h1 id="消息顺序性"><a href="#消息顺序性" class="headerlink" title="消息顺序性"></a>消息顺序性</h1><p>流式消息必然存在消息顺序性的问题，在 <code>ClientCall.java</code> 中提到</p><figure class="highlight java"><table><tbody><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br></pre></td><td class="code"><pre><span class="line"> <span class="comment">/**</span></span><br><span class="line"><span class="comment"> * Requests up to the given number of messages from the call to be delivered to</span></span><br><span class="line"><span class="comment"> * {<span class="doctag">@link</span> Listener#onMessage(Object)}. No additional messages will be delivered.</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * &lt;p&gt;Message delivery is guaranteed to be sequential in the order received. In addition, the</span></span><br><span class="line"><span class="comment"> * listener methods will not be accessed concurrently. While it is not guaranteed that the same</span></span><br><span class="line"><span class="comment"> * thread will always be used, it is guaranteed that only a single thread will access the listener</span></span><br><span class="line"><span class="comment"> * at a time.</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * &lt;p&gt;If it is desired to bypass inbound flow control, a very large number of messages can be</span></span><br><span class="line"><span class="comment"> * specified (e.g. {<span class="doctag">@link</span> Integer#MAX_VALUE}).</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * &lt;p&gt;If called multiple times, the number of messages able to delivered will be the sum of the</span></span><br><span class="line"><span class="comment"> * calls.</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * &lt;p&gt;This method is safe to call from multiple threads without external synchronization.</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@param</span> numMessages the requested number of messages to be delivered to the listener. Must be</span></span><br><span class="line"><span class="comment"> *                    non-negative.</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@throws</span> IllegalStateException if call is not {<span class="doctag">@code</span> start()}ed</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@throws</span> IllegalArgumentException if numMessages is negative</span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">abstract</span> <span class="keyword">void</span> <span class="title">request</span><span class="params">(<span class="keyword">int</span> numMessages)</span></span>;</span><br></pre></td></tr></tbody></table></figure><p>由此可得 gRPC 遵守消息到达顺序。</p><h1 id="流式消息中的背压（不存在）"><a href="#流式消息中的背压（不存在）" class="headerlink" title="流式消息中的背压（不存在）"></a>流式消息中的背压（不存在）</h1><p>流式消息必然涉及到这样的问题，当请求发送速度远大于服务端对请求处理速度时，持续的请求可能会压垮服务端。这时可以阻塞请求，来达到降低请求发送速度的目的，可称为背压。但 gRPC 中没有背压。？？？有点诧异。</p><h2 id="非阻塞"><a href="#非阻塞" class="headerlink" title="非阻塞"></a>非阻塞</h2><p>但这并不是说 gRPC 没有处理这种问题的能力。首先确认下客户端发送请求是否有可能阻塞。一路跟踪代码下探到 transport 层，在 Stream.java#writeMessage(msg) 的注释中提到：</p><blockquote><p><em>This method will always return immediately and will not wait for the write to complete.</em></p></blockquote><p>故客户端发送是不会阻塞的。</p><h2 id="流控制"><a href="#流控制" class="headerlink" title="流控制"></a>流控制</h2><p>但是 gRPC 是基于 Http2 的，Http2 有流控的机制，简单来说，接收端可以给发送端设定一个窗口值。以此，可以限制客户端发送的速度，但是没有背压就意味着没法限制用户，这样可能导致客户端的待发送缓存爆掉，问题还是没法解决。</p><h2 id="onReady"><a href="#onReady" class="headerlink" title="onReady"></a>onReady</h2><p>别着急，看下 <code>CallStreamObserver.java</code></p><figure class="highlight java"><table><tbody><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * If {<span class="doctag">@code</span> true}, indicates that the observer is capable of sending additional messages</span></span><br><span class="line"><span class="comment"> * without requiring excessive buffering internally. This value is just a suggestion and the</span></span><br><span class="line"><span class="comment"> * application is free to ignore it, however doing so may result in excessive buffering within the</span></span><br><span class="line"><span class="comment"> * observer.</span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">abstract</span> <span class="keyword">boolean</span> <span class="title">isReady</span><span class="params">()</span></span>;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * Set a {<span class="doctag">@link</span> Runnable} that will be executed every time the stream {<span class="doctag">@link</span> #isReady()} state</span></span><br><span class="line"><span class="comment"> * changes from {<span class="doctag">@code</span> false} to {<span class="doctag">@code</span> true}.  While it is not guaranteed that the same</span></span><br><span class="line"><span class="comment"> * thread will always be used to execute the {<span class="doctag">@link</span> Runnable}, it is guaranteed that executions</span></span><br><span class="line"><span class="comment"> * are serialized with calls to the 'inbound' {<span class="doctag">@link</span> StreamObserver}.</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * &lt;p&gt;On client-side this method may only be called during {<span class="doctag">@link</span></span></span><br><span class="line"><span class="comment"> * ClientResponseObserver#beforeStart}. On server-side it may only be called during the initial</span></span><br><span class="line"><span class="comment"> * call to the application, before the service returns its {<span class="doctag">@code</span> StreamObserver}.</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * &lt;p&gt;Note that the handler may be called some time after {<span class="doctag">@link</span> #isReady} has transitioned to</span></span><br><span class="line"><span class="comment"> * true as other callbacks may still be executing in the 'inbound' observer.</span></span><br><span class="line"><span class="comment"> *</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@param</span> onReadyHandler to call when peer is ready to receive more messages.</span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">abstract</span> <span class="keyword">void</span> <span class="title">setOnReadyHandler</span><span class="params">(Runnable onReadyHandler)</span></span>;</span><br></pre></td></tr></tbody></table></figure><p>可看到作为替代，用户可以使用 onReady 的监听，这样可以避免待发送消息爆掉。</p><h2 id="小结"><a href="#小结" class="headerlink" title="小结"></a>小结</h2><p>gRPC 这个方案看似比背压更复杂，但实际上更合理。首先依靠 Http2 从网络上控制请求的频率，将异常和问题拦截在客户端，锅分得合理。然后在客户端，提供了解决问题的方案，并允许设定 handler，对客户端来说也是简单方便的。</p><h1 id="参考"><a href="#参考" class="headerlink" title="参考"></a>参考</h1><ol><li><a href="https://github.com/grpc/grpc-java" target="_blank" rel="noopener">gRpc-java Readme</a>.</li><li><a href="https://www.nihaoshijie.com.cn/index.php/archives/698/" target="_blank" rel="noopener">HTTP2.0关于多路复用的研究</a></li></ol></div><footer class="post-footer"><div class="post-tags"> <a href="/tags/源码分析/" rel="tag"># 源码分析</a> <a href="/tags/gRPC/" rel="tag"># gRPC</a></div><div class="post-nav"><div class="post-nav-next post-nav-item"><a href="/开发/7c949937.html" rel="next" title="Envoy 统计数据子系统源码分析"><i class="fa fa-chevron-left"></i> Envoy 统计数据子系统源码分析</a></div><span class="post-nav-divider"></span><div class="post-nav-prev post-nav-item"> <a href="/开发/5b9f8a4a.html" rel="prev" title="Envoy service mesh、Prometheus和Grafana下的微服务监控（翻译）">Envoy service mesh、Prometheus和Grafana下的微服务监控（翻译）<i class="fa fa-chevron-right"></i></a></div></div></footer></div></article></div></div><div class="comments" id="comments"></div></div><div class="sidebar-toggle"><div class="sidebar-toggle-line-wrap"><span class="sidebar-toggle-line sidebar-toggle-line-first"></span><span class="sidebar-toggle-line sidebar-toggle-line-middle"></span><span class="sidebar-toggle-line sidebar-toggle-line-last"></span></div></div><aside id="sidebar" class="sidebar"><div class="sidebar-inner"><ul class="sidebar-nav motion-element"><li class="sidebar-nav-toc sidebar-nav-active" data-target="post-toc-wrap"> 文章目录</li><li class="sidebar-nav-overview" data-target="site-overview-wrap"> 站点概览</li></ul><div class="site-overview-wrap sidebar-panel"><div class="site-overview"><div class="site-author motion-element" itemprop="author" itemscope="" itemtype="http://schema.org/Person"> <img class="site-author-image" itemprop="image" src="/images/blogAvatar.jpg" alt="moeyui"><p class="site-author-name" itemprop="name">moeyui</p><p class="site-description motion-element" itemprop="description">moeyui | moeyui's blog</p></div><nav class="site-state motion-element"><div class="site-state-item site-state-posts"> <a href="/archives/"><span class="site-state-item-count">39</span> <span class="site-state-item-name">日志</span></a></div><div class="site-state-item site-state-categories"> <a href="/categories/index.html"><span class="site-state-item-count">6</span> <span class="site-state-item-name">分类</span></a></div><div class="site-state-item site-state-tags"> <a href="/tags/index.html"><span class="site-state-item-count">36</span> <span class="site-state-item-name">标签</span></a></div></nav><div class="feed-link motion-element"><a href="/atom.xml" rel="alternate"><i class="fa fa-rss"></i> RSS</a></div><div class="links-of-author motion-element"><span class="links-of-author-item"><a href="https://github.com/moeyui1" title="GitHub → https://github.com/moeyui1" rel="noopener" target="_blank"><i class="fa fa-fw fa-github"></i> GitHub</a></span><span class="links-of-author-item"><a href="https://twitter.com/moeyui1" title="Twitter → https://twitter.com/moeyui1" rel="noopener" target="_blank"><i class="fa fa-fw fa-twitter"></i> Twitter</a></span></div></div></div><div class="post-toc-wrap motion-element sidebar-panel sidebar-panel-active"><div class="post-toc"><div class="post-toc-content"><ol class="nav"><li class="nav-item nav-level-1"><a class="nav-link" href="#Overview"><span class="nav-number">1.</span> <span class="nav-text">Overview</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#Stub"><span class="nav-number">1.1.</span> <span class="nav-text">Stub</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#Channel"><span class="nav-number">1.2.</span> <span class="nav-text">Channel</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#Transport"><span class="nav-number">1.3.</span> <span class="nav-text">Transport</span></a></li></ol></li><li class="nav-item nav-level-1"><a class="nav-link" href="#使用"><span class="nav-number">2.</span> <span class="nav-text">使用</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#客户端"><span class="nav-number">2.1.</span> <span class="nav-text">客户端</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#服务端"><span class="nav-number">2.2.</span> <span class="nav-text">服务端</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#StreamObserver"><span class="nav-number">2.3.</span> <span class="nav-text">StreamObserver</span></a></li></ol></li><li class="nav-item nav-level-1"><a class="nav-link" href="#简图"><span class="nav-number">3.</span> <span class="nav-text">简图</span></a></li><li class="nav-item nav-level-1"><a class="nav-link" href="#分层分析"><span class="nav-number">4.</span> <span class="nav-text">分层分析</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#Stub-层"><span class="nav-number">4.1.</span> <span class="nav-text">Stub 层</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#Channel-层"><span class="nav-number">4.2.</span> <span class="nav-text">Channel 层</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#Transport-层"><span class="nav-number">4.3.</span> <span class="nav-text">Transport 层</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#协议层"><span class="nav-number">4.4.</span> <span class="nav-text">协议层</span></a></li></ol></li><li class="nav-item nav-level-1"><a class="nav-link" href="#消息顺序性"><span class="nav-number">5.</span> <span class="nav-text">消息顺序性</span></a></li><li class="nav-item nav-level-1"><a class="nav-link" href="#流式消息中的背压（不存在）"><span class="nav-number">6.</span> <span class="nav-text">流式消息中的背压（不存在）</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#非阻塞"><span class="nav-number">6.1.</span> <span class="nav-text">非阻塞</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#流控制"><span class="nav-number">6.2.</span> <span class="nav-text">流控制</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#onReady"><span class="nav-number">6.3.</span> <span class="nav-text">onReady</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#小结"><span class="nav-number">6.4.</span> <span class="nav-text">小结</span></a></li></ol></li><li class="nav-item nav-level-1"><a class="nav-link" href="#参考"><span class="nav-number">7.</span> <span class="nav-text">参考</span></a></li></ol></div></div></div></div></aside></div></main><footer id="footer" class="footer"><div class="footer-inner"><div class="copyright">© 2015 – <span itemprop="copyrightYear">2020</span><span class="with-love" id="animate"><i class="fa fa-user"></i></span> <span class="author" itemprop="copyrightHolder">moeyui</span></div><div class="powered-by">由 <a href="https://hexo.io" class="theme-link" rel="noopener" target="_blank">Hexo</a> 强力驱动 v3.9.0</div> <span class="post-meta-divider">|</span><div class="theme-info">主题 – <a href="https://theme-next.org" class="theme-link" rel="noopener" target="_blank">NexT.Gemini</a> v6.7.0</div></div></footer><div class="back-to-top"><i class="fa fa-arrow-up"></i></div></div><script src="//cdn1.lncld.net/static/js/3.11.1/av-min.js"></script><script src="//unpkg.com/valine/dist/Valine.min.js"></script><style>.copy-btn{display:inline-block;padding:6px 12px;font-size:13px;font-weight:700;line-height:20px;color:#333;white-space:nowrap;vertical-align:middle;cursor:pointer;background-color:#eee;background-image:linear-gradient(#fcfcfc,#eee);border:1px solid #d5d5d5;border-radius:3px;user-select:none;outline:0}.highlight-wrap .copy-btn{transition:opacity .3s ease-in-out;opacity:0;padding:2px 6px;position:absolute;right:4px;top:8px}.highlight-wrap .copy-btn:focus,.highlight-wrap:hover .copy-btn{opacity:1}.highlight-wrap{position:relative}</style><script src="/bundle.js"></script><script>"[object Function]"!==Object.prototype.toString.call(window.Promise)&&(window.Promise=null);var GUEST=["nick","mail","link"],guest="nick,mail,link";guest=guest.split(",").filter(function(e){return-1<GUEST.indexOf(e)}),new Valine({el:"#comments",verify:!1,notify:!1,appId:"Nk9tNzbllEl79DVtRiOuLmbm-gzGzoHsz",appKey:"uPzxSfSzF95Gav6vGx2dOhCq",placeholder:"Just go go",avatar:"retro",meta:guest,pageSize:"10",visitor:!0}),$(".highlight").each(function(e,t){var n=$("<div>").addClass("highlight-wrap");$(t).after(n),n.append($("<button>").addClass("copy-btn").append("复制").on("click",function(e){var t=$(this).parent().find(".code").find(".line").map(function(e,t){return $(t).text()}).toArray().join("\n"),n=document.createElement("textarea"),o=document.createRange(),i=window.getSelection(),a=window.pageYOffset||document.documentElement.scrollTop;n.style.top=a+"px",n.style.position="absolute",n.style.opacity="0",n.value=t,n.textContent=t,n.contentEditable=!0,n.readOnly=!1,document.body.appendChild(n),o.selectNode(n),i.removeAllRanges(),i.addRange(o),n.setSelectionRange(0,t.length);document.execCommand("copy");n.blur(),$(this).blur()})).on("mouseleave",function(e){var t=$(this).find(".copy-btn");setTimeout(function(){t.text("复制")},300)}).append(t)})</script></body></html>